package org.codehaus.activemq.message.util;

import EDU.oswego.cs.dl.util.concurrent.SynchronizedInt;
import java.io.File;
import java.io.IOException;
import java.util.List;
import javax.jms.JMSException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.message.DefaultWireFormat;
import org.codehaus.activemq.message.Packet;
import org.codehaus.activemq.message.WireFormat;

public class SpooledBoundedPacketQueue
  implements BoundedPacketQueue
{
  private String name;
  private DataContainer container;
  private WireFormat wireFormat;
  private long maxDataLength;
  private boolean closed;
  private boolean stopped;
  private SynchronizedInt size = new SynchronizedInt(0);
  private Object inLock = new Object();
  private Object outLock = new Object();
  private static int WAIT_TIMEOUT = 250;
  private static final Log log = LogFactory.getLog(SpooledBoundedPacketQueue.class);

  public SpooledBoundedPacketQueue(File dir, String name, long maxDataLength, int maxBlockSize)
    throws IOException
  {
    char[] chars = name.toCharArray();
    for (int i = 0; i < chars.length; i++) {
      if (!Character.isLetterOrDigit(chars[i])) {
        chars[i] = '_';
      }
    }
    this.name = new String(chars);
    this.maxDataLength = maxDataLength;
    this.wireFormat = new DefaultWireFormat();
    this.container = new DataContainer(dir, this.name, maxBlockSize);

    this.container.deleteAll();
  }

  public SpooledBoundedPacketQueue(File dir, String name)
    throws IOException
  {
    this(dir, name, 67108864L, 8192);
  }

  public void enqueue(Packet packet)
    throws JMSException
  {
    if (!isFull()) {
      enqueueNoBlock(packet);
    }
    else {
      synchronized (this.inLock) {
        try {
          while (isFull())
            this.inLock.wait(WAIT_TIMEOUT);
        }
        catch (InterruptedException ie)
        {
        }
      }
      enqueueNoBlock(packet);
    }
  }

  public void enqueueNoBlock(Packet packet)
    throws JMSException
  {
    try
    {
      byte[] data = this.wireFormat.toBytes(packet);
      this.size.increment();
      this.container.write(data);
    }
    catch (IOException e) {
      JMSException jmsEx = new JMSException("toBytes failed");
      jmsEx.setLinkedException(e);
      throw jmsEx;
    }
    synchronized (this.outLock)
    {
      byte[] data;
      this.outLock.notify();
    }
  }

  public Packet dequeue()
    throws JMSException, InterruptedException
  {
    Packet result = null;
    synchronized (this.outLock) {
      while ((result = dequeueNoWait()) == null) {
        this.outLock.wait(WAIT_TIMEOUT);
      }
    }
    return result;
  }

  public Packet dequeue(long timeInMillis)
    throws JMSException, InterruptedException
  {
    Packet result = dequeueNoWait();
    if (result == null) {
      synchronized (this.outLock) {
        this.outLock.wait(timeInMillis);
        result = dequeueNoWait();
      }
    }
    return result;
  }

  public Packet dequeueNoWait()
    throws JMSException, InterruptedException
  {
    Packet result = null;
    if (this.stopped) {
      synchronized (this.outLock) {
        while ((this.stopped) && (!this.closed)) {
          this.outLock.wait(WAIT_TIMEOUT);
        }
      }
    }
    try
    {
      byte[] data = this.container.read();
      if (data != null) {
        result = this.wireFormat.fromBytes(data);
        this.size.decrement();
      }
    }
    catch (IOException e) {
      JMSException jmsEx = new JMSException("fromBytes failed");
      jmsEx.setLinkedException(e);
      throw jmsEx;
    }
    byte[] data;
    if ((result != null) && (!isFull())) {
      synchronized (this.inLock) {
        this.inLock.notify();
      }
    }
    return result;
  }

  public boolean isFull()
  {
    return this.container.length() >= this.maxDataLength;
  }

  public void close()
  {
    try
    {
      this.closed = true;
      this.container.close();
    }
    catch (IOException ioe) {
      log.warn("Couldn't close queue", ioe);
    }
  }

  public String getName()
  {
    return this.name;
  }

  public int size()
  {
    return this.size.get();
  }

  public boolean isStarted()
  {
    return !this.stopped;
  }

  public void stop()
  {
    synchronized (this.outLock) {
      this.stopped = true;
    }
  }

  public void start()
  {
    this.stopped = false;
    synchronized (this.outLock) {
      this.outLock.notifyAll();
    }
    synchronized (this.inLock) {
      this.inLock.notifyAll();
    }
  }

  public boolean isEmpty()
  {
    return this.size.get() == 0;
  }

  public void clear()
  {
  }

  public List getContents()
  {
    return null;
  }
}